package com.example.verticles;

import io.vertx.core.AbstractVerticle;
import io.vertx.core.CompositeFuture;
import io.vertx.core.Future;
import io.vertx.core.Promise;
import io.vertx.rabbitmq.RabbitMQClient;
import io.vertx.rabbitmq.RabbitMQConsumer;
import io.vertx.rabbitmq.RabbitMQOptions;

public class RabbitVerticle extends AbstractVerticle {

    private RabbitMQClient client;

    @Override
    public void start() throws Exception {
        RabbitMQOptions config = new RabbitMQOptions();
// Each parameter is optional
// The default parameter with be used if the parameter is not set
        config.setUser("admin");
        config.setPassword("admin");
        config.setHost("101.133.174.208");
        config.setPort(5672);
        config.setVirtualHost("/");
        config.setConnectionTimeout(6000); // in milliseconds
        config.setRequestedHeartbeat(60); // in seconds
        config.setHandshakeTimeout(6000); // in milliseconds
        config.setRequestedChannelMax(5);
        config.setNetworkRecoveryInterval(500); // in milliseconds
        config.setAutomaticRecoveryEnabled(true);

        client=RabbitMQClient.create(vertx, config);

        method01();
    }

    private void method03() {
        Promise<Void> connectPromise = Promise.promise();
        client.start(connectPromise);


//        CompositeFuture.all()
    }

    private void method02() {

        client.exchangeDeclare("CNY_CMDS","topic",true,false,exchange->{
            if (exchange.succeeded()){
                client.queueDeclare("CNY_CMDS_vertx",true,false,false,queue->{
                    if (queue.succeeded()){
                        client.queueBind("CNY_CMDS_vertx","CNY_CMDS","CNY_CMDS-1",bind->{
                            if (bind.succeeded()){
                                client.basicConsumer("CNY_CMDS_vertx", rabbitMQConsumerAsyncResult -> {
                                    if (rabbitMQConsumerAsyncResult.succeeded()) {
                                        System.out.println("RabbitMQ consumer created !");
                                        RabbitMQConsumer mqConsumer = rabbitMQConsumerAsyncResult.result();
                                        mqConsumer.handler(message -> {
                                            System.out.println(Thread.currentThread().getName()+" Got message: "
                                                    + message.body().toString());
                                        });
                                    } else {
                                        rabbitMQConsumerAsyncResult.cause().printStackTrace();
                                    }
                                });
                            }else {
                                bind.cause().printStackTrace();
                            }
                        });
                    }else {
                        queue.cause().printStackTrace();
                    }
                });
            }else {
                exchange.cause().printStackTrace();
            }
        });
    }

    private void method01() {
        client.start(asyncResult -> {
            if (asyncResult.succeeded()) {
                System.out.println("RabbitMQ successfully connected!");
                method02();
            } else {
                System.out.println("Fail to connect to RabbitMQ " + asyncResult.cause().getMessage());
            }
        });
    }
}
